package service

import (
	"bytes"
	"encoding/base64"
	"eznat/core"
	"eznat/utils"
	"fmt"
	"log"
	"net"
	"strings"
	"sync"
	"time"
)

var locker sync.RWMutex
var instance *ChannelService
var once sync.Once

const(
	ON = 1
	OFF = 0
	INTERRUPT = 2
)

type ChannelService struct {
	Conn     net.Conn
	EventBus chan interface{}

	ip      string
	port    string
	dataBus string

	convert *utils.Converter
	conf    *core.Config
	ServiceStatus int

	ChannelServiceStatus int

	localServer map[string] *bytes.Buffer
	packageLengthByte [4]byte
	readTimeOut time.Duration
}

func NewChannelService() *ChannelService {
	this := &ChannelService{}
	this.conf = core.NewConfig()
	this.readTimeOut = time.Millisecond * 1000
	this.ip = this.conf.ChannelIp
	this.port = this.conf.ChannelPort
	this.dataBus = this.conf.DataBus

	this.localServer = make(map[string] *bytes.Buffer)
	this.EventBus = make(chan interface{})
	return this
}

/**
 * 设备上线数据包发送
 */
func (myself *ChannelService) clientOnline() {
	for {
		if myself.ServiceStatus == ON {
			data := make(map[string]interface{})
			data["data_bus"] = myself.conf.GetDataBus()
			myself.Publish(core.EVENT_IN_CLIENT_ONLINE, data)
			time.Sleep(time.Millisecond * 1500)
		} else {
			break
		}
	}
}

func (myself *ChannelService) Connect() {
	if myself.ServiceStatus == ON {
		log.Println("正在连接通道服务器...")
		conn, err := net.Dial("tcp", fmt.Sprintf("%s:%s", myself.ip, myself.port))
		if err != nil {
			log.Println(" 通道服务器连接失败，正在重连", err)
			myself.Connect()
			time.Sleep(time.Second * 1)
			return
		}
		myself.Conn = conn
		tcpConn, _ := conn.(*net.TCPConn)
		tcpConn.SetNoDelay(true)
		tcpConn.SetReadBuffer(10 * 1024 * 1024)
		tcpConn.SetWriteBuffer(10 * 1024 * 1024)
		myself.ChannelServiceStatus = ON
		myself.Subscribe(core.EVENT_OUT_CONNECT + myself.dataBus)
		myself.Subscribe(core.EVENT_OUT_MSG +  myself.dataBus)
		myself.Subscribe(core.EVENT_OUT_CLOSE +  myself.dataBus)
		go myself.clientOnline()
		log.Println("通道服务器连接成功...")
	}
}
func (myself *ChannelService) Start() {
	myself.ServiceStatus = ON
	myself.Connect()
	for {
		if myself.ServiceStatus == ON {
			// log.Println("循环接受数据")
			n, connBroken := myself.Conn.Read(myself.packageLengthByte[:])

			if connBroken != nil && n == 0 {
				timeOut :=  strings.Contains(connBroken.Error(), "i/o timeout")
				if  timeOut {
					continue
				} else {
					myself.ChannelServiceStatus = INTERRUPT
					myself.Connect()
				}
			}

			packageLength := myself.convert.BytesToInt(myself.packageLengthByte[:])
			if packageLength > 65535 {
				continue
			}

			onePackage := myself.ReadOnePackage(packageLength)
			receive, err := myself.convert.UnSerialize(string(onePackage)).(map[string]interface{})
			if err != true {
				log.Println("收到错误数据：", string(onePackage), connBroken)
				continue
			}

			data := receive["data"].(map[string]interface{})
			channel := data[core.CHANNEL].(string)
			event := receive["channel"].(string)
			localServer := myself.localServer
			_, exist := localServer[channel]
			if !exist && event == core.EVENT_OUT_CONNECT + myself.dataBus {
				localServer[channel] = new(bytes.Buffer)
				go myself.shunt(channel) // 数据分流
				rBuffer := append(myself.convert.IntToBytes(len(onePackage)), onePackage...)
				localServer[channel].Write(rBuffer)
				continue
			}
			if exist {
				rBuffer := append(myself.convert.IntToBytes(len(onePackage)), onePackage...)
				localServer[channel].Write(rBuffer)
			}
		} else if myself.ServiceStatus == OFF {
			break
		}
	}
}
/**
 * 对输入的数据进行分流处理，输入的数据混杂在一起，用 bytes.Buffer 分流。
 */
func (myself *ChannelService) shunt( channelKey string) {
	localConnection := make(map[string]net.Conn)
	bufferIO := myself.localServer[channelKey]
	waitWrite := time.Millisecond * 10 // 如果没有数据，睡眠一下，让出 bufferIO，不处于一直读取状态，否则无法写入
	exit := false
	defer func() {
		sendData := make(map[string]interface{})
		sendData[core.CHANNEL] = channelKey
		myself.Publish(core.EVENT_IN_CLOSE + myself.dataBus, sendData)
		localConnection[channelKey].Close()
		delete( myself.localServer, channelKey)
		log.Println("退出本地服务协程")
	}()

	for {
		time.Sleep(waitWrite)
		if exit == true {
			break
		}

		var lengthByte [4]byte
		n, _ := bufferIO.Read(lengthByte[:]) // 非阻塞，可能没有数据。所以需要判断一下
		if n < 4 {
			continue
		}

		len := myself.convert.BytesToInt(lengthByte[:])
		buf := bufferIO.Next(len)
		receiveTmp := myself.convert.UnSerialize(string(buf))
		var receive map[string]interface{}
		if receiveTmp != nil {
			receive = receiveTmp.(map[string]interface{})
		} else {
			continue
		}
		data := receive["data"].(map[string]interface{})
		event := receive["channel"].(string)
		channel := data[core.CHANNEL].(string)

		if event == core.EVENT_OUT_CONNECT + myself.dataBus {
			uri := data[core.URI].(string)
			log.Println("EV_OUT_CONNECT", channel, uri)
			var err error
			localConnection[channel], err = net.Dial("tcp", uri)
			if err != nil {
				fmt.Println("err :", err)
				return
			}
			go func() {
				receiveBuffer := make([]byte, 65535)
				for {
					localConnection[channel].SetReadDeadline(time.Now().Add(time.Second * 10))
					dataLen, err := localConnection[channel].Read(receiveBuffer[:]) // 阻塞
					if err != nil {
						log.Println("EV_IN_CLOSE.", err)
						exit = true
						break
					}
					buffer := receiveBuffer[:dataLen]
					if dataLen > 0 {
						sendData := make(map[string]interface{})
						sendData[core.CHANNEL] = channel
						sendData[core.DATA] = base64.StdEncoding.EncodeToString(buffer)
						myself.Publish(core.EVENT_IN_MSG + myself.dataBus, sendData)

						// log.Println("EV_IN_MSG")
					}
				}
			}()
		} else if event == core.EVENT_OUT_MSG + myself.dataBus {
			// log.Println("EV_OUT_MSG", channel, data)
			receiveData, _ := base64.StdEncoding.DecodeString(data[core.DATA].(string))
			_, ok := localConnection[channel]
			if ok {
				localConnection[channel].Write(receiveData)
			}
		}
	}
}
func (myself *ChannelService) Stop() {
	myself.ServiceStatus = OFF
	myself.ChannelServiceStatus = OFF
}
func (myself *ChannelService) Status() interface{}{
	return myself.ServiceStatus
}
func (myself *ChannelService) ReadOnePackage(packageLength int) []byte {
	const receiveBufferSize = 65535
	alreadyReceive := 4
	var rBuffer []byte
	var tmpBuffer [receiveBufferSize]byte
	for {
		if alreadyReceive < packageLength {
			n, _ := myself.Conn.Read(tmpBuffer[:packageLength-alreadyReceive])
			alreadyReceive += n
			rBuffer = append(rBuffer, tmpBuffer[:n]...)
			continue
		} else {
			// fmt.Println("收到完整数据包:", alreadyReceive)
			break
		}
	}
	return rBuffer
}

func (myself *ChannelService) Publish(eventName string, data map[string]interface{}) {
	sendDataString := myself.packData(eventName, "publish", data)
	myself.send(sendDataString)
}
func (myself *ChannelService) Subscribe(eventName string) {
	sendDataString := myself.packData(eventName, "subscribe", make(map[string]interface{}))
	myself.send(sendDataString)
}
func (myself *ChannelService) Unsubscribe(eventName string) {
	sendDataString := myself.packData(eventName, "unsubscribe", make(map[string]interface{}))
	myself.send(sendDataString)
}
func (myself *ChannelService) send(data string) {
	length := myself.convert.IntToBytes(4 + len(data))
	writeData := append(length, []byte(data)...)
	myself.Conn.Write(writeData)
}
func (myself *ChannelService) packData(eventName string, types string, data map[string]interface{}) string {
	sendData := make(map[string]interface{})
	channels := make(map[int]string)
	channels[0] = eventName
	sendData["type"] = types
	sendData["channels"] = channels
	sendData["data"] = data
	sendDataString := myself.convert.Serialize(sendData)
	return sendDataString
}
